package cn.middle.remote.server;

import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import cn.middle.remote.message.BaseSignal;
import cn.middle.remote.util.NettyAttributeUtil;
import cn.middle.remote.util.PrioritySort;
import io.netty.channel.Channel;

public class NettySession implements Session {
	
	private final Logger log = LoggerFactory.getLogger(NettySession.class);
	
	private volatile AtomicBoolean writeStatus = new AtomicBoolean(false); //默认状态为flase 
	
	@SuppressWarnings("rawtypes")
	private PriorityBlockingQueue<PrioritySort> queue = new PriorityBlockingQueue<>(100);

	
	private volatile State targetState = State.ACTIVE;
	
	enum State {
		ACTIVE,DLE;
	}

	private Channel channel;
	
	
	public NettySession(Channel channel) {
		this.channel = channel;
	}
	
	@Override
	public int clientId() {
		return NettyAttributeUtil.clientId(channel);
	}

	@Override
	public synchronized <T extends BaseSignal> void send(T signal) {
		if(State.DLE == targetState) {
			return ;
		}
		queue.add(new PrioritySort<>(signal));
		submit();
	}
	
	
	public void submit() {
		if(writeStatus.compareAndSet(false, true)) {  //写状态为flase时 创建线程并更新状态为true，并创建发送线程，循环发送数据
			new Thread(()->{
				boolean runState = true;
				while(runState) {
					try {
						PrioritySort<?> signal = queue.poll();
						if(signal == null) { //当队列为空时，更改写入状态，并停止当前线程节省线程消耗
							writeStatus.compareAndSet(true, false);
							runState = false;  
							return;
						}
						channel.writeAndFlush(signal.getValue());
					}catch(Throwable t) {
						writeStatus.compareAndSet(true, false);
						runState = false;  
						log.error(t.getMessage(),t);
					}
				}
			}).start();
		}
	}

	@Override
	public synchronized void close() {
		targetState = State.DLE;
		queue.clear();
		writeStatus.compareAndSet(true, false);
	}

}
